package com.sskj.base.utils;

/*
 * 作者 :吕志豪
 * 简书：https://www.jianshu.com/u/6e525b929aac
 * github：https://github.com/lvzhihao100
 * 描述：
 * 创建时间：2018-07-26 11:31
 */


import org.reactivestreams.Subscriber;

import java.security.cert.X509Certificate;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;

import io.reactivex.Flowable;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;
import okio.ByteString;


public abstract class WebSocketObserver extends Flowable<String> {
    private String wsUrl;
    private OkHttpClient okHttpClient;
    private Request request;
    private Subscriber<? super String> observer;
    private boolean isClose;
    private WebSocket mWebSocket;
    private String msg;
    private Disposable timerDispo;

    public WebSocketObserver(String wsUrl) {
        this.wsUrl = wsUrl;

    }

    public abstract void onOpen(WebSocket webSocket, Response response, String type);

    @Override
    protected void subscribeActual(Subscriber<? super String> subscriber) {
        this.observer = subscriber;
        connect();

    }

    private void setSSL(OkHttpClient.Builder builder) throws Exception {
        X509TrustManager x509TrustManager = new X509TrustManager() {
            @Override
            public void checkClientTrusted(
                    X509Certificate[] chain,
                    String authType) {
            }

            @Override
            public void checkServerTrusted(
                    X509Certificate[] chain,
                    String authType) {
            }

            @Override
            public X509Certificate[] getAcceptedIssuers() {
                return new X509Certificate[0];
            }
        };
        TrustManager[] trustAllCerts = new TrustManager[]{x509TrustManager};

        // Install the all-trusting trust manager
        final SSLContext sslContext = SSLContext.getInstance("SSL");
        sslContext.init(null, trustAllCerts, new java.security.SecureRandom());
        HostnameVerifier hostnameVerifier = (hostname, session) -> true;
        builder.sslSocketFactory(sslContext.getSocketFactory(), x509TrustManager)
                .hostnameVerifier(hostnameVerifier);
    }

    private void connect() {
        isClose = false;
        DisposUtil.close(timerDispo);
        OkHttpClient.Builder builder = new OkHttpClient.Builder();
        try {
            setSSL(builder);
        } catch (Exception e) {
            e.printStackTrace();
        }
        okHttpClient = builder
                .readTimeout(13, TimeUnit.SECONDS).build();
        request = new Request.Builder()
                .url(wsUrl)
                .build();
        mWebSocket = okHttpClient.newWebSocket(request, new WebSocketListener() {
            @Override
            public void onOpen(WebSocket webSocket, Response response) {
                super.onOpen(webSocket, response);
                System.out.println("onOpen");
                WebSocketObserver.this.onOpen(webSocket, response, msg);
                DisposUtil.close(timerDispo);
                timerDispo = Flowable.interval(5, TimeUnit.SECONDS)
                        .subscribe(aLong -> {
                            if (mWebSocket!=null&&!isClose) {
                                mWebSocket.send("ping");
                            }
                        });
            }

            @Override
            public void onMessage(WebSocket webSocket, String text) {
                super.onMessage(webSocket, text);
                observer.onNext(text);
            }

            @Override
            public void onMessage(WebSocket webSocket, ByteString bytes) {
                super.onMessage(webSocket, bytes);
                observer.onNext(bytes.utf8());

            }

            @Override
            public void onClosing(WebSocket webSocket, int code, String reason) {
                super.onClosing(webSocket, code, reason);
                webSocket.close(1001, "onClosing");
                System.out.println("onClosing" + reason);
                if (!isClose) {
                    reConnect();
                }
            }

            @Override
            public void onFailure(WebSocket webSocket, Throwable t,  Response response) {
                super.onFailure(webSocket, t, response);
                webSocket.close(1001, "onFailure");
                System.out.println("onFailure" + t.toString());
                if (!isClose) {

                    reConnect();
                }

            }

            private void reConnect() {
                System.out.println("15秒后重新连接。。。");

                Flowable.timer(15, TimeUnit.SECONDS)
                        .subscribeOn(Schedulers.newThread())
                        .subscribe(i -> {
                            System.out.println("重新连接中。。。");
                            connect();
                        }, e -> {
                            System.out.println("15秒连接出错" + e.toString());
                        });
            }

            @Override
            public void onClosed(WebSocket webSocket, int code, String reason) {
                super.onClosed(webSocket, code, reason);
                System.out.println("onClosed" + reason);
                isClose = true;
                if (code != 1000 && !"close by me".equals(reason)) {
                    reConnect();
                }
            }
        });
    }

    public void send(String msg) {
        if (mWebSocket != null) {
            this.msg = msg;

            mWebSocket.send(msg);
        }
    }

    public void close() {
        if (mWebSocket != null) {
            isClose = true;
            System.out.println("onDispose 关闭了");
            DisposUtil.close(timerDispo);
            mWebSocket.close(1000, "close by me");
            mWebSocket = null;
        }
    }


}